package com.tpvlog.seckill.common;

import redis.clients.jedis.Jedis;

import java.util.*;

/**
 * 自己封装的用于进行redis集群数据分片的工具类
 */
public class RedisCluster {

    /**
     * Redis集群对应的Jedis列表
     */
    private List<Jedis> cluster = new ArrayList<Jedis>();

    /**
     * 私有构造函数
     */
    private RedisCluster() {
        // 初始化redis集群
        cluster.add(new Jedis("127.0.0.1", 6479));
        cluster.add(new Jedis("127.0.0.1", 6579));
        cluster.add(new Jedis("127.0.0.1", 6679));
    }

    /**
     * 单例
     */
    private static class Singleton {

        static RedisCluster instance = new RedisCluster();

    }

    /**
     * 获取单例
     * @return
     */
    public static RedisCluster getInstance() {
        return Singleton.instance;
    }

    /**
     * 初始化秒杀商品库存
     * @param productId 商品id
     * @param seckillStock 秒杀库存
     */
    public void initSeckillProductStock(Long productId, Long seckillStock) {
        // 计算每个redis节点上的库存分片数量
        int clusterSize = cluster.size();
        Long seckillStockPerNode = seckillStock / clusterSize;
        Long remainSeckillStock = seckillStock - seckillStockPerNode * clusterSize;
        Long seckillStockLastNode = seckillStockPerNode + remainSeckillStock;

        System.out.println("每个redis节点（除最后一个节点外）的库存数量为：" + seckillStockPerNode);
        System.out.println("最后一个redis节点的库存数量为：" + seckillStockLastNode);
        System.out.println("redis集群的总库存数量为：" + (seckillStockPerNode * (clusterSize - 1) + seckillStockLastNode));

        // 对除最后一个redis节点之外的其他节点，进行库存分片数据初始化
        for(int i = 0; i < cluster.size() - 1; i++) {
            Jedis jedis = cluster.get(i);
            jedis.hset("seckill::product::" + productId + "::stock", "salingStock", String.valueOf(seckillStockPerNode));
            jedis.hset("seckill::product::" + productId + "::stock", "lockedStock", "0");
            jedis.hset("seckill::product::" + productId + "::stock", "saledStock", "0");
        }

        // 对redis集群最后一个节点，进行库存分片数据初始化
        Jedis jedis = cluster.get(cluster.size() - 1);
        jedis.hset("seckill::product::" + productId + "::stock", "salingStock", String.valueOf(seckillStockLastNode));
        jedis.hset("seckill::product::" + productId + "::stock", "lockedStock", "0");
        jedis.hset("seckill::product::" + productId + "::stock", "saledStock", "0");
    }

    /**
     * 基于redis集群进行秒杀抢购
     * @param productId
     */
    public Boolean flashSale(Long userId, Long productId) {
        // 随机选择一个Redis的节点
        int redisNodeCount = cluster.size();
        int chosenRedisNodeIndex = new Random().nextInt(redisNodeCount);
        Jedis chosenRedisNode = cluster.get(chosenRedisNodeIndex);

        // 向redis节点提交一个lua脚本进行抢购
        String flashSaleLuaScript = ""
                + "local productKey = 'seckill::product::" + productId + "::stock';"
                + "local salingStock = redis.call('hget', productKey, 'salingStock') + 0;"
                + "local lockedStock = redis.call('hget', productKey, 'lockedStock') + 0;"
                + "if(salingStock > 0) "
                + "then "
                    + "redis.call('hset', productKey, 'salingStock', salingStock - 1);"
                    + "redis.call('hset', productKey, 'lockedStock', lockedStock + 1);"
                    + "return 'success';"
                + "else "
                    + "return 'fail';"
                + "end;";

        String flashSaleResult = null;

        try {
            flashSaleResult = (String) chosenRedisNode.eval(flashSaleLuaScript);
        } catch(Exception e) {
            // 如果这里报错了，那么很有可能就是某一台redis机器崩溃了
            // 崩溃之后的处理逻辑就在这里写就可以了
            // 只不过是一部分的库存分片不可用了，但是此时你可以去找其他的库存分片来进行秒杀
            // 跑一下下面的库存分片迁移的逻辑，尝试去其他机器上进行秒杀就可以了
            try {
                return tryOtherStockShard(chosenRedisNodeIndex, flashSaleLuaScript, userId, productId);
            } catch(Exception e1) {
                // 在这里就可以写所有的redis节点都崩溃的逻辑了
                // 就可以尝试把请求给写入到本地磁盘去，让抢购状态保持在抢购中
                return false;
            }
        }

        // 如果秒杀抢购成功了
        if("success".equals(flashSaleResult)) {
            JedisManager jedisManager = JedisManager.getInstance();
            Jedis jedis = jedisManager.getJedis();
            jedis.set("flash_sale::stock_shard::" + userId + "::" + productId,
                    String.valueOf(chosenRedisNodeIndex));
            return true;
        }
        // 如果第一次秒杀抢购失败了，则进行库存分片迁移的操作
        else {
            try {
                return tryOtherStockShard(chosenRedisNodeIndex, flashSaleLuaScript, userId, productId);
            } catch(Exception e) {
                // 在这里就可以写所有的redis节点都崩溃的逻辑了
                // 就可以尝试把请求给写入到本地磁盘去，让抢购状态保持在抢购中
                return false;
            }
        }
    }

    /**
     * 尝试其他的库存分片节点
     * @param failedStockShard
     * @param flashSaleLuaScript
     * @param userId
     * @param productId
     * @return
     */
    private Boolean tryOtherStockShard(int failedStockShard,
                                       String flashSaleLuaScript,
                                       Long userId,
                                       Long productId) throws Exception {
        String flashSaleResult = null;
        Boolean flashSaleSuccess = false;
        Boolean allRedisNodeCrashed = true;

        for(int i = 0; i < cluster.size(); i++) {
            if(i != failedStockShard) {
                try {
                    Jedis redisNode = cluster.get(i);

                    flashSaleResult = (String) redisNode.eval(flashSaleLuaScript);
                    allRedisNodeCrashed = false;

                    if("success".equals(flashSaleResult)) {
                        JedisManager jedisManager = JedisManager.getInstance();
                        Jedis jedis = jedisManager.getJedis();
                        jedis.set("flash_sale::stock_shard::" + userId + "::" + productId,
                                String.valueOf(i));

                        flashSaleSuccess = true;
                        break;
                    }
                } catch(Exception e) {
                    // 在尝试其他节点进行抢购的时候，其他某个节点也出现了宕机问题
                }
            }
        }

        // 如果说所有的redis节点都崩溃了
        if(allRedisNodeCrashed) {
            throw new Exception("所有Redis节点都崩溃了！！！");
        }

        return flashSaleSuccess;
    }

    /**
     * 秒杀订单支付成功的库存处理逻辑
     * @param stockShardRedisNode
     * @param productId
     */
    public void flashSaleOrderPaySuccess(String stockShardRedisNode, Long productId) {
        Integer redisNodeIndex = Integer.valueOf(stockShardRedisNode);
        Jedis redisNode = cluster.get(redisNodeIndex);

        String luaScript = ""
                + "local productKey = 'seckill::product::" + productId + "::stock';"
                + "local lockedStock = redis.call('hget', productKey, 'lockedStock') + 0;"
                + "local saledStock = redis.call('hget', productKey, 'saledStock') + 0;"
                + "redis.call('hset', productKey, 'lockedStock', lockedStock - 1);"
                + "redis.call('hset', productKey, 'saledStock', lockedStock + 1);";

        redisNode.eval(luaScript);
    }

    /**
     * 秒杀订单支付失败的库存处理逻辑
     * @param stockShardRedisNode
     * @param productId
     */
    public void flashSaleOrderPayFail(String stockShardRedisNode, Long productId) {
        Integer redisNodeIndex = Integer.valueOf(stockShardRedisNode);
        Jedis redisNode = cluster.get(redisNodeIndex);

        String luaScript = ""
                + "local productKey = 'seckill::product::" + productId + "::stock';"
                + "local salingStock = redis.call('hget', productKey, 'salingStock') + 0;"
                + "local lockedStock = redis.call('hget', productKey, 'lockedStock') + 0;"
                + "redis.call('hset', productKey, 'lockedStock', lockedStock - 1);"
                + "redis.call('hset', productKey, 'salingStock', salingStock + 1);";

        redisNode.eval(luaScript);
    }

}
